package promote.brozen.algorithm.cainiao;

import java.io.*;
import java.nio.CharBuffer;
import java.nio.MappedByteBuffer;
import java.nio.channels.FileChannel;
import java.nio.charset.StandardCharsets;
import java.nio.file.Paths;
import java.nio.file.StandardOpenOption;
import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.LongAdder;

/**
 *
 * 在机器上字符乱码问题还没解决，暂时还不能运行。
 *
 * 思路是把一个文件拆成N块分别统计
 * 每一块的第一行和最后一行可能是个半行，因此tail要和下一块的head拼接起来再统计，最后一块的tail单独统计
 *
 * @author Brozen
 * @date 2020/9/30 9:14 PM
 * @email brozen@qq.com
 */
public class FindErrorCode {

    // 并发线程数量
    int concurrency = 1;

    private ConcurrentHashMap<Integer, LongAdder> codeCounts;

    private ExecutorService pool;

    public FindErrorCode() {
        this.codeCounts = new ConcurrentHashMap<>();
        this.pool = new ThreadPoolExecutor(concurrency, concurrency, 10, TimeUnit.SECONDS,
                new LinkedBlockingQueue<>());
    }


    public void count() throws IOException, ExecutionException, InterruptedException {

        FileChannel fc = FileChannel.open(Paths.get("/data/logs/error.log"), StandardOpenOption.READ);
        long fileSize = fc.size();
        long stepLength = fileSize / concurrency;
        // utf8 char=16bit
        stepLength = stepLength % 2 == 0 ? stepLength : (stepLength + 1L);

        List<Future<CountResult>> futures = new ArrayList<>((int) ((fileSize / stepLength) + 1));
        for (long i = 0; i < fileSize;) {
            long end = i + stepLength > fileSize ? fileSize : i + stepLength;
            long length = end - i;

            // 生成N个任务，每个任务统计stepLength长度
            MappedByteBuffer buffer = fc.map(FileChannel.MapMode.READ_ONLY, i, length);
            futures.add(pool.submit(new CountTask(buffer)));

            i += stepLength;
            if (length != stepLength) {
                break;
            }
        }

        CountResult lastResult = null;
        for (Future<CountResult> future : futures) {
            CountResult result = future.get();
            if (lastResult != null) {
                String code = lastResult.tail + result.head;
                if (!code.isEmpty()) {
                    Integer c = Integer.valueOf(code);
                    LongAdder adder = codeCounts.computeIfAbsent(c, cd -> new LongAdder());
                    adder.add(1);
                }
            }

            lastResult = result;
        }

        // 最后一个tail
        if (lastResult != null && !lastResult.tail.isEmpty()) {
            Integer c = Integer.valueOf(lastResult.tail);
            LongAdder adder = codeCounts.computeIfAbsent(c, cd -> new LongAdder());
            adder.add(1);
        }

        int firstCode = -1;
        long firstCount = 0;
        int secondCode = -1;
        long secondCount = 0;
        int thirdCode = -1;
        long thirdCount = 0;

        int swapCode;
        long swapCount;
        for (Map.Entry<Integer, LongAdder> entry : codeCounts.entrySet()) {
            int code = entry.getKey();
            long count = entry.getValue().longValue();


            if (count > firstCount) {
                swapCode = firstCode;
                firstCode = code;
                code = swapCode;

                swapCount = firstCount;
                firstCount = count;
                count = swapCount;
            }

            if (count > secondCount) {
                swapCode = secondCode;
                secondCode = code;
                code = swapCode;

                swapCount = secondCount;
                secondCount = count;
                count = swapCount;
            }

            if (count > thirdCount) {
                thirdCode = code;
                thirdCount = count;
            }
        }

        System.out.println(String.format("first[%s=%s] second[%s=%s] third[%s=%s]",
                firstCode, firstCount, secondCode, secondCount, thirdCode, thirdCount));

        pool.shutdownNow();
    }


    class CountTask implements Callable<CountResult> {

        private CharBuffer buffer;

        public CountTask(MappedByteBuffer buffer) {
            CharBuffer cbuffer = CharBuffer.allocate(buffer.limit());
            StandardCharsets.UTF_8.newDecoder().decode(buffer, cbuffer, true);
            this.buffer = (CharBuffer) cbuffer.flip();
        }

        @Override
        public CountResult call() {

            boolean firstLineRecorded = false;
            StringBuilder line = new StringBuilder();
            CountResult result = new CountResult();
            while (buffer.hasRemaining()) {
                char c = buffer.get();
                if (c == '\n') {
                    // 第一个记录在head
                    if (!firstLineRecorded) {
                        result.head = line.toString();
                        firstLineRecorded = true;
                    } else {
                        Integer code = Integer.valueOf(line.toString());
                        LongAdder adder = codeCounts.computeIfAbsent(code, cd -> new LongAdder());
                        adder.add(1L);
                    }

                    line.setLength(0);
                } else {
                    line.append(c);
                }
            }

            // 最后一个记录在tail
            if (line.length() > 0) {
                result.tail = line.toString();
            }

            return result;
        }
    }

    class CountResult {
        String head = "";
        String tail = "";
    }


    public static void main(String[] args) throws IOException, ExecutionException, InterruptedException {
        long ts = System.currentTimeMillis();
        new FindErrorCode().count();

        System.out.println("cost: " + (System.currentTimeMillis() - ts) + "ms");
    }


}
